package io.mvvm.ws;

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @program: Netty
 * @description: 通道管理以及消息广播
 * @author: 潘
 * @create: 2021-04-23 20:31
 **/
public final class WebSocketChannelHolder {
    /**
     * KEY：用户ID
     * VALUE：用户所在的 Channel
     */
    private static final Map<String, Channel> CHANNEL_MAP = new ConcurrentHashMap<>();
    /**
     * 所有的 Channel
     * 参数：全局事件执行器
     */
    private static final ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 添加通道
     *
     * @param userId  channel key
     * @param channel channel
     * @return true or false
     */
    public static boolean addChannel(String userId, Channel channel) {
        CHANNEL_MAP.put(userId, channel);
        return CHANNEL_GROUP.add(channel);
    }

    /**
     * 判断通道是否已经注册
     *
     * @param userId channelKey
     * @return true or false
     */
    public static boolean isRegister(String userId) {
        Channel channel = CHANNEL_MAP.get(userId);
        if (null == channel) {
            return false;
        }

        // 如果通道非活动状态则去除
        if (!channel.isActive()) {
            removeChannel(userId);
            return false;
        }

        return true;
    }

    public static String isRegister(Channel channel) {
        if (CHANNEL_MAP.containsValue(channel)) {
            Iterator<Map.Entry<String, Channel>> entryIt = CHANNEL_MAP.entrySet().iterator();
            while (entryIt.hasNext()) {
                Map.Entry<String, Channel> entry = entryIt.next();
                if (entry.getValue().equals(channel)) {
                    return entry.getKey();
                }
            }
        }
        return "";
    }
    /**
     * 按照通道KEY删除指定通道
     *
     * @param userId channelKey
     * @return true or false
     */
    public static boolean removeChannel(String userId) {
        Channel channel = CHANNEL_MAP.remove(userId);
        boolean result = CHANNEL_GROUP.remove(channel);
        channel.close();
        return result;
    }

    /**
     * 删除通道
     *
     * @param channel
     * @return
     */
    public static String removeChannel(Channel channel) {
        if (CHANNEL_MAP.containsValue(channel)) {
            Iterator<Map.Entry<String, Channel>> entryIt = CHANNEL_MAP.entrySet().iterator();
            while (entryIt.hasNext()) {
                Map.Entry<String, Channel> entry = entryIt.next();
                if (entry.getValue().equals(channel)) {
                    entryIt.remove();
                    CHANNEL_GROUP.remove(channel);
                    channel.close();
                    return entry.getKey();
                }
            }
        }
        return "";
    }

    /**
     * 刷新用户 webSocket 连接通道
     * 用户连接通道只存在一个新来的连接接提原来的
     *
     * @param userId
     * @param channel
     */
    public static void refreshChannel(String userId, Channel channel) {
        if (null == channel) {
            return;
        }

        Channel channelOld = CHANNEL_MAP.get(userId);
        if (channelOld.equals(channel)) {
            return;
        }

        CHANNEL_MAP.put(userId, channel);
        CHANNEL_GROUP.add(channel);
        CHANNEL_GROUP.remove(channelOld);
        channelOld.flush();
        channelOld.close();
    }

    /**
     * 给指定通道发送消息
     *
     * @param userId
     * @param msg
     * @return
     */
    public static boolean sendMsg(String userId, String msg) {
        return sendMsg(Collections.singletonList(userId), msg);
    }

    /**
     * 广播消息
     *
     * @param userId
     * @param msg
     * @return
     */
    public static boolean sendMsg(List<String> userId, String msg) {
        List<Channel> cs = userId.stream()
                .map(CHANNEL_MAP::get)
                .filter(ch -> null != ch && ch.isActive())
                .collect(Collectors.toList());
        ChannelGroupFuture channelFutures = CHANNEL_GROUP.writeAndFlush(new TextWebSocketFrame(msg), cs::contains);
        return channelFutures.isSuccess();
    }

}
